package com.myrabbitmq.mylistener.returnlistener;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

public class ReturnProductor {
    public static void main(String[] args) {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("129.211.131.113");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        try {
            Connection connection = connectionFactory.newConnection();

            Channel channel = connection.createChannel();

//            //指定消息确认模式
//            channel.confirmSelect();


            Map<String ,Object> headers = new HashMap<>();
            headers.put("bizName","1111");
            headers.put("activeName","222");
            headers.put("taskDispatchServiceName","222");

            AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
                    .deliveryMode(2) //持久化消息为2，这样可以保证重启了消息依然存在
                    .contentEncoding("UTF-8")
                    .expiration("10000") //10秒没有被消费，那么将会被自动移除
                    .headers(headers)
                    .build();
            String exchangeName = "test_consumer_exchange";
            String routingKey = "test.consumer.save";
//            String routingKey = "test.consumer.key2";
//            String routingKey3 = "test.topic.key3";

                //发送数据
                String msg = "Hello RabbitMQ message return" ;
                channel.basicPublish(exchangeName,routingKey,true,properties,(msg).getBytes());
//                channel.basicPublish(exchangeName,routingKey2,null,(msg+routingKey2).getBytes());
//                channel.basicPublish(exchangeName,routingKey3,null,(msg+routingKey3).getBytes());

            //6、添加一个确认监听
            channel.addConfirmListener(new ConfirmListener() {
                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    System.err.println("=========== ack =========");

                }

                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    System.err.println("===========no ack =======");
                }
            });

            //7、增加返回监听
            channel.addReturnListener(new ReturnListener() {
                @Override
                public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.err.println("===========返回失败=======");
                    System.err.println("replyCode:"+replyCode);
                    System.err.println("replyText:"+replyText);
                    System.err.println("exchange:"+exchange);
                    System.err.println("routingKey:"+routingKey);
                    System.err.println("properties:"+properties);
                    System.err.println("body:"+new String(body));
                }
            });

//            //关闭连接
//            channel.close();
//            connection.close();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}
